package com.epoint.flinkdemo.window;

import com.epoint.flinkdemo.entity.SenSorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

public class TimeWindowMinTempTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度为1
        env.setParallelism(1);

        // 设置时间语义为事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // socket source
        DataStreamSource<String> source = env.socketTextStream("localhost", 7777);

        SingleOutputStreamOperator<SenSorReading> mapSource = source.map(new MapFunction<String, SenSorReading>() {
            @Override
            public SenSorReading map(String s) throws Exception {
                String[] split = s.split(",");
                return new SenSorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
            }
        });

        // 设置放慢3秒的watermark
        SingleOutputStreamOperator<SenSorReading> waterSource = mapSource.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SenSorReading>(Time.seconds(3)) {
            @Override
            public long extractTimestamp(SenSorReading element) {
                return element.getTimestamp() * 1000;
            }
        });

        // 开窗要设置keyBy
        SingleOutputStreamOperator<SenSorReading> resultSource = waterSource.keyBy("id")
                .timeWindow(Time.seconds(15))
                .allowedLateness(Time.seconds(30))
                .min("temp");

        resultSource.print("minTemp===>");
        env.execute();
    }
}
